package ithink.link.stack;

import com.codahale.metrics.Counter;
import com.codahale.metrics.Meter;
import io.netty.buffer.PooledByteBufAllocator;
import ithink.link.base.influxdb.Influxdb;
import ithink.link.base.message.LinkMessageDecoder;
import ithink.link.base.message.LinkMessageEncoder;
import ithink.link.base.message.LinkMessage;
import ithink.link.base.utils.InfluxdbHelper;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import ithink.link.ConfigHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Created by zaqb on 17-9-11.
 */
public class LinkStack {
    private static final Logger LOGGER = LoggerFactory.getLogger(LinkStack.class);

    private final ConcurrentHashMap<String,LinkConnection> inConnections = new ConcurrentHashMap<>();

    private String name;
    private boolean inited = false;
    private boolean started = false;
    private String fixedKey = "ITHI";
    private int  readTimeoutMs = 0;

    private IListener listener = null;

    private EventLoopGroup acceptGroup = new NioEventLoopGroup();
    private EventLoopGroup readwriteGroup = new NioEventLoopGroup();

    private final Object syncCon = new Object();

    private Counter counterOfConnection;
    private Meter meterOfConnected;
    private Meter meterOfDisconnected;

    private Meter meterOfMsg;

    private Meter meterOfError;

    public LinkStack(String name) {
        this.name = name;
        counterOfConnection = InfluxdbHelper.counter(null, name + ".link.connection.count");
        meterOfConnected = InfluxdbHelper.meter(null, name + ".link.connect");
        meterOfDisconnected =  InfluxdbHelper.meter(null, name + ".link.disconnect");
        meterOfMsg = InfluxdbHelper.meter(null, name + ".link.msg");
        meterOfError = InfluxdbHelper.meter(null, name + ".link.msg");
    }

    public IListener getListener() {
        return listener;
    }

    public void setListener(IListener listener) {
        this.listener = listener;
    }

    public String getFixedKey() {
        return fixedKey;
    }

    public void setFixedKey(String fixedKey) {
        this.fixedKey = fixedKey;
    }

    public void init(Config config) {
        if(inited) {
            throw new RuntimeException("failed to init");
        }

        readTimeoutMs = config.readTimeoutMs;
        acceptGroup = new NioEventLoopGroup(config.acceptGroupCount);
        readwriteGroup = new NioEventLoopGroup(config.rwGroupCount);

        LOGGER.info("stack {} init with {}", name, config);
        inited = true;

    }

    public synchronized void start(int port) throws Exception {
        if(!inited) {
            throw new RuntimeException("not init");
        }
        if(started) {
            throw new Exception("has started");
        }
        try {
            if(name == null|| name.isEmpty()) {
                name = "port."+port;
            }

            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(acceptGroup, readwriteGroup).channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 2048)
                    //Boss线程内存池配置.
                    .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    //Work线程内存池配置.
                    .childOption(ChannelOption.ALLOCATOR,PooledByteBufAllocator.DEFAULT)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            LinkStack.this.initIncomingChannel(socketChannel);
                        }
                    });
            ChannelFuture f = serverBootstrap.bind(port).sync();
            if(!f.isSuccess())
            {
                throw new Exception(String.format("failed to listener on port %d",port));
            }
            LOGGER.info("link stack {} start on {}", name, port);
            started = true;
        }
        catch (Exception ex)
        {
            LOGGER.error("link stack {} start failed {}", name, ex);
            throw ex;
        }
    }


    public synchronized void stop() {
        //TODO stop gracefully;
    }

    void onMessageReceived(LinkConnection connection, LinkMessage message)
    {
        LOGGER.debug("message receive from {}:{}, {}",
                connection.remoteIp(),
                connection.remotePort(),
                message.getText().toJSONString());
        meterOfMsg.mark();
        if(listener!=null)
        {
            listener.onMessageReceived(connection,message);
        }
        else {
            LOGGER.warn("unhandle message from connection {}:{}", connection.remoteIp(), connection.remotePort());
        }
    }

    void exceptionCaught(Exception ex) {
        meterOfError.mark();
        LOGGER.error("caught exception: {}", ex);
    }

    void onMessageSendFailed(LinkConnection connection,LinkMessage message)
    {
        if(listener!=null)
        {
            listener.onMessageSendFailed(connection,message);
        }
        else {
            LOGGER.warn("unhandle message send failed from connection {}:{}", connection.remoteIp(), connection.remotePort());
        }
    }
    public void initIncomingChannel(SocketChannel socketChannel) {
        if(readTimeoutMs>0)
        {
            socketChannel.pipeline().addLast(new IdleStateHandler(readTimeoutMs,0,0, TimeUnit.MILLISECONDS));
        }
        LinkMessageDecoder msgDecoder = new LinkMessageDecoder();
        LinkMessageEncoder msgEncoder = new LinkMessageEncoder();
        msgDecoder.setFixedKey( fixedKey );
        msgEncoder.setFixedKey( fixedKey );
        socketChannel.pipeline().addLast(msgDecoder).addLast( msgEncoder );
        socketChannel.pipeline().addLast(new LinkConnection(LinkStack.this, msgDecoder, msgEncoder));

    }


    void registerConnection(LinkConnection connection)
    {
        LOGGER.info("connection {}:{} register to stack", connection.remoteIp(), connection.remotePort());
        meterOfConnected.mark();
       // connection.setContext( this.name );
        String hashKey = getConnectionKey(connection.remoteIp(),connection.remotePort());
        boolean hasPut;
        synchronized (syncCon)
        {
            LinkConnection con = inConnections.putIfAbsent(hashKey,connection);
            LOGGER.info("current inConnections size = {}", inConnections.size());
            hasPut = (con == null);
        }

        if(!hasPut) {
            LOGGER.error("failed to registerConnection,connection {}:{}", connection.remoteIp(), connection.remotePort());
            connection.disconnect();
        }
        else {
            counterOfConnection.inc();
            listener.onConnected(connection);

        }
    }

    void unregisterConnection(LinkConnection connection)
    {
        LOGGER.info("connection {}:{} unregister from stack", connection.remoteIp(), connection.remotePort());
        meterOfDisconnected.mark();
        synchronized (syncCon) {
            String hashKey = getConnectionKey(connection.remoteIp(),connection.remotePort());
            if (!inConnections.containsKey(hashKey)) {
                LOGGER.error("failed to unregisterConnection,connection {}:{}", connection.remoteIp(), connection.remotePort());
                return;
            }
            inConnections.remove(hashKey);
            LOGGER.info("current inConnections size = {}", inConnections.size());
            counterOfConnection.dec();
        }

        listener.onDisconnected(connection);

    }



    private String getConnectionKey(String ip,int port)
    {
        return String.format("%s:%d",ip,port);
    }

    public interface IListener
    {
        void onConnected(IConnection connection);
        void onDisconnected(IConnection connection);
        void onMessageReceived(IConnection connection, LinkMessage message);
        void onMessageSendFailed(IConnection connection, LinkMessage message);

    }
    public static class Config
    {
        int readTimeoutMs = 30*1000;
        int acceptGroupCount = 1;
        int rwGroupCount = 4;

        public Config(int readTimeoutMs, int keepAliveIntervalMs, int acceptGroupCount, int rwGroupCount) {
            this.readTimeoutMs = readTimeoutMs;
            this.acceptGroupCount = acceptGroupCount;
            this.rwGroupCount = rwGroupCount;
        }

        public Config() {
            readTimeoutMs = ConfigHelper.getConnectionTTL()*1000;
        }

        @Override
        public String toString() {
            return "Config{" +
                    "readTimeoutMs=" + readTimeoutMs +
                    ", acceptGroupCount=" + acceptGroupCount +
                    ", rwGroupCount=" + rwGroupCount +
                    '}';
        }
    }

}
